package com.katze.common.ws.handler;

import com.katze.common.ws.core.MessageDispatcher;
import com.katze.common.ws.core.UserChannelGroup;
import com.katze.common.ws.protocol.Message;
import com.katze.common.ws.protocol.MessageType;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;

public class MessageHandler extends SimpleChannelInboundHandler<Message> {
    private static final Logger log = LoggerFactory.getLogger(MessageHandler.class);

    private final MessageDispatcher dispatcher;
    private final UserChannelGroup channels = new UserChannelGroup();

    public MessageHandler(MessageDispatcher dispatcher) {
        this.dispatcher = dispatcher;
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Message message) throws Exception {
        Channel channel = ctx.channel();
        if (message.getType().equals(MessageType.AUTH)) {
            try {
                JwtAuthHandler handler = dispatcher.getJwtAuthHandler();
                Object data = handler.doAuthentication((String) message.getData(), user -> {
                    if (user.equals(message.getSender())) {
                        channels.bind(user, channel);
                    } else {
                        throw new IllegalAccessError("非法Token");
                    }
                });
                if (data != null) {
                    channel.writeAndFlush(message.setData(data)).addListener(handler);
                }
            } catch (Throwable e) {
                if (e instanceof IllegalAccessError) {
                    message.setCode(401);
                } else {
                    message.setCode(500);
                }
                message.setData(e.getMessage());
                channel.writeAndFlush(message).addListeners(ChannelFutureListener.CLOSE);
            }
        } else {
            if (channels.isBound(channel)) {
                dispatcher.dispatch(channel, message);
            } else {
                message.notification("限制访问", "消息通道尚未认证,禁止连接").error();
                channel.writeAndFlush(message).addListeners(ChannelFutureListener.CLOSE);
            }
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        dispatcher.close(ctx.channel());
        super.channelInactive(ctx);
    }

    /**
     * exception 异常 Caught 抓住 抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        Serializable user = ctx.channel().attr(UserChannelGroup.USER).get();
        log.error("{}({})连接出错", user, ctx.channel().id(), cause);
        ctx.close();
    }
}
